package com.shenjies88.practice.reactor.wrapper;

import com.shenjies88.practice.reactor.thread.WorkRunnable;

import java.nio.channels.SelectionKey;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author shenjies88
 * @since 2022/12/7/23:42
 */
public class WorkWrapper {

    private final int index;

    private final int WORK_SIZE = 5;
    private final int WORK_MAX_SIZE = 20;
    private final LinkedBlockingQueue<SelectionKey> queue = new LinkedBlockingQueue<>();
    private final ThreadPoolExecutor WORK_POOL = new ThreadPoolExecutor(
            WORK_SIZE,
            WORK_MAX_SIZE,
            30,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1),
            new ThreadFactory() {
                private final AtomicInteger threadNumber = new AtomicInteger(1);

                @Override
                public Thread newThread(Runnable r) {
                    String name = String.format("work-%s-thread-%s", index, threadNumber.getAndIncrement());
                    return new Thread(null, r, name);
                }
            },
            new ThreadPoolExecutor.CallerRunsPolicy());

    public WorkWrapper(int index) {
        this.index = index;
    }

    public void init(int i) {
        new Thread(() -> {
            while (true) {
                SelectionKey poll = null;
                try {
                    poll = queue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    System.err.println("工作线程池出队异常");
                }
                if (poll != null) {
                    WORK_POOL.submit(new WorkRunnable(poll));
                }

            }
        }, "work-pool-consume-" + i).start();
    }

    public void queue(SelectionKey selectionKey) {
        try {
            queue.put(selectionKey);
        } catch (InterruptedException e) {
            e.printStackTrace();
            System.err.println("投递工作线程池队列异常");
        }
    }
}
